package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/8/28 18:20
 */
public class Datagen2Kafka_GeomData {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        BIGINT NOT NULL,\n" +
                "    sensor_name        STRING,\n" +
                "    tem        DOUBLE,\n" +
                "    ts   TIMESTAMP(3)" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '5', \n" +
                "  'rows-per-second' = '1', \n" +
                "  'fields.sensor_name.length' = '5', \n" +
                "  'fields.id.min' = '1', \n" +
                "  'fields.id.max' = '10000', \n" +
                "  'fields.tem.min' = '0', \n" +
                "  'fields.tem.max' = '100'\n" +
                ")");

//        //Sink
        tableEnv.executeSql("CREATE TABLE sinkTable (" +
                "    id        BIGINT NOT NULL,\n" +
                "    sensor_name        STRING,\n" +
                "    tem        DOUBLE,\n" +
                "    ts   TIMESTAMP(3)" +
                "   ) " +
                "   WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'ods_test_sensor',\n" +
                "  'properties.bootstrap.servers' = '192.168.102.154:29092',\n" +
                "  'format' = 'json'\n" +               //json格式序列化:{"id":1,"sid":"11baac5a","ts":"2023-06-12 15:09:22.849","flag":true,"vc":31}
                ")");

//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        //Insert
        tableEnv.executeSql("insert into sinkTable() select * from sourceTable");

    }
}
